﻿# -*- coding: utf-8 -*-
from utils.Lock import ReentrantRedLock
from concurrent.futures import ThreadPoolExecutor
from setting import stream_name, groupname, consumername_profix, Prefixs
import threading
from decorator import with_app_context
import time
from utils.def_errors import SqlExecError, Error
from exts import db
from blueprints.common import VoucherHelper
from utils.redis import Redis

pool = ThreadPoolExecutor(max_workers=None)  # 不指定最大线程,默认为cpu线程数*5
r = Redis.r
try:
    # 查看所有消费者组
    groups_info = r.xinfo_groups(stream_name)
    # 如果当前消费者组不存在就创建
    if not groups_info:
        # id为0默认从消息队列头开始读，为$默认只读最新的消息 ,mkstream为True: 不存在流就创建
        r.xgroup_create(name=stream_name, groupname=groupname, id='0', mkstream=True)
    else:
        group_names = [group_info['name'].decode('utf-8') for group_info in groups_info]
        if groupname not in group_names:
            r.xgroup_create(name=stream_name, groupname=groupname, id='0', mkstream=True)
except Exception as e:
    # 报错说明没有stream，依然是创建消费者组，自动创建stream
    r.xgroup_create(name=stream_name, groupname=groupname, id='0', mkstream=True)


@with_app_context
def creat_order(userId: int, voucherID: int, order_id: int):
    try:
        # 创建一个分布式锁实例
        key = Prefixs.lock_voucher_order_prefix + str(userId)
        distlock = ReentrantRedLock(key, ttl=5000)
        # 尝试去拿到锁资源，拿不到返回错误
        if not distlock.tryLock():
            print('获取锁失败,服务器异常！')
            raise Error("获取锁失败！")
        # 创建订单
        voucher_order = VoucherHelper.voucher_order_create(userID=userId, voucherID=voucherID,
                                                           order_id=order_id)
        db.session.add(voucher_order)
        db.session.commit()
    except Exception as e:
        print(f"订单 order_id:{order_id},userId:{userId},voucherID:{voucherID} 创建失败", e)
        db.session.rollback()
        raise SqlExecError("订单sql创建失败！")
    finally:
        distlock.unLock()  # 即使前面return了也会执行


# 循环执行定时任务
def creat_order_task(threadId):
    while True:
        # 一直循环从消息队里拿到消息进行处理
        try:
            # 从消费者组中读数据
            result = r.xreadgroup(groupname, consumername_profix + str(threadId), {stream_name: '>'}, count=1,
                                  block=2000)
            if not result or not result[0][1]:  # 有时候有result，但是result[0][1]为空列表
                continue
            messages_info = result[0][1]
            msg_id = messages_info[0][0].decode('utf-8')
            message = messages_info[0][1]
            message = {key.decode('utf-8'): message[key].decode('utf-8') for key in message}
            creat_order(int(message['userId']), int(message['voucherId']), int(message['id']))
            r.xack(stream_name, groupname, msg_id)
        except Exception as e:
            print(e)
            # 如果捕获到异常，代表有一条消息没有处理成功，就尝试拿到一条没有应答的消息
            while True:
                try:
                    result = r.xreadgroup("g1", 'c1', {"stream.order": '0'}, count=1, block=2000)
                    # 如果拿不到没有应答的消息，代表消息处理完毕
                    if not result or not result[0][1]:  # 有时候有result，但是result[0][1]为空列表
                        break
                    messages_info = result[0][1]
                    msg_id = messages_info[0][0].decode('utf-8')
                    message = {key.decode('utf-8'): message[key].decode('utf-8') for key in messages_info[0][1]}
                    creat_order(message['userId'], message['voucherId'], message['order_id'])
                    r.xack(stream_name, groupname, msg_id)
                except Exception as e:
                    print(e)
                    time.sleep(0.2)

                # 开启3线程做异步循环任务，读取消息队列信息，完成订单信息的库存任务


for _ in range(3):
    # 都在同一个消费者组，用threadId区分身份
    pool.submit(creat_order_task, threading.get_ident())
